package com.javaspeak.concurrency.nio.niolet.server;

import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.LockSupport;

import org.apache.log4j.Logger;

/**
 * @author John Dickerson
 */
public class MessageResponseThread extends Thread {

    public Logger logger = Logger.getLogger( MessageResponseThread.class );

    private ConcurrentLinkedQueue<Message> messageResponseConcurrentLinkedQueue;
    private Map<SelectionKey, ByteBuffer> readBufferBySelectionKeyMap;
    private volatile boolean keepRunning = true;


    /**
     * Remove any internal state for the given key.
     * @param key the key to be cancelled.
     */
    private void resetKey( SelectionKey selectionKey ) {

        selectionKey.cancel();
        readBufferBySelectionKeyMap.remove( selectionKey );
    }


    /**
     * Callback method for when the server disconnects an client connection.
     * @param key the key for the disconnected client.
     */
    public void disconnected( SelectionKey selectionKey ){

        logger.debug( "Disconnecting client" );
    }


    private void writeResponseMessage( Message message ){

        ByteBuffer byteBuffer = message.getMessageByteBuffer();
        SelectionKey selectionKey = message.getSelectionKey();

        int sizeOfMessage = byteBuffer.limit();

        ByteBuffer writeBuffer = ByteBuffer.allocate( sizeOfMessage + 4 );
        writeBuffer.putInt( sizeOfMessage );
        writeBuffer.put( byteBuffer );

        writeBuffer.flip();

        if ( keepRunning ) {

            int bytesWritten;

            try {
                 // only 1 thread can write to a channel at a time
                SocketChannel channel = ( SocketChannel )selectionKey.channel();

                synchronized ( channel ) {

                    bytesWritten = channel.write( writeBuffer );
                }

                if ( bytesWritten==-1 ) {
                    resetKey( selectionKey );
                    disconnected( selectionKey );
                }
            } catch ( Throwable e ) {
                resetKey( selectionKey );
                disconnected( selectionKey );
            }
        }
    }


    /**
     * Constructor
     *
     * @param messageResponseConcurrentLinkedQueue
     */
    public MessageResponseThread(
            ConcurrentLinkedQueue<Message> messageResponseConcurrentLinkedQueue,
                Map<SelectionKey, ByteBuffer> readBufferBySelectionKeyMap ){

        this.messageResponseConcurrentLinkedQueue =
            messageResponseConcurrentLinkedQueue;

        this.readBufferBySelectionKeyMap = readBufferBySelectionKeyMap;
    }


    public void gracefullyStop(){

        this.keepRunning = false;
    }


    public void run() {

        Message message;

        while ( keepRunning ){

            message = messageResponseConcurrentLinkedQueue.poll();

            if ( message == null ){

                LockSupport.park();
            }
            else {

                writeResponseMessage( message );
            }
        }
    }
}
